package com.boot.rxjava;

import io.reactivex.Observable;
import io.reactivex.observables.ConnectableObservable;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class MainTest {
    public static void main(String[] args) throws InterruptedException {
//        User user = new User();
//        user.username = "tony";
//        user.addresses = new ArrayList<>();
//        Address address1 = new Address("ren ming road", "su zhou");
//        Address address2 = new Address("dong wu bei road", "su zhou");
//        user.addresses.add(address1);
//        user.addresses.add(address2);
//        Observable.just(user).map(u -> u.addresses).subscribe(System.out::println);
//        Observable.just(user).flatMap(u -> Observable.fromIterable(u.addresses)).subscribe(System.out::println);
//        Observable.just(user).concatMap(u -> Observable.fromIterable(u.addresses)).subscribe(System.out::println);
//        Observable.range(0, 10).buffer(3).subscribe(System.out::println);
//        System.out.println("----------------------------");
//        Observable.range(0, 10).buffer(5, 5)
//                .subscribe(System.out::println, System.out::println, () -> System.out.println("Completed"));
//        System.out.println("----------------------------");
//        Observable.range(0, 10).window(4).subscribe(integerObservable ->{
//            System.out.println("onNext:");
//            integerObservable.subscribe(System.out::println, System.out::println, () -> System.out.println("completed"));
//        });
//        System.out.println("----------------------------");
//        Observable.just(1, 2, 3).first(100).subscribe(System.out::println);
//        Observable.empty().first(100).subscribe(System.out::println);
//        Observable.just(1, 2, 3).firstElement().subscribe(System.out::println);
//        Observable.empty().firstOrError().subscribe(System.out::println);
//        Observable.just(1, 2, 3).last(100).subscribe(System.out::println);
//        Observable.empty().last(100).subscribe(System.out::println);
//        Observable.range(0, 4).take(3).subscribe(System.out::println);
//        Observable.range(0, 4).take(7).subscribe(System.out::println);
//        System.out.println("----------------------------");
//        Observable.intervalRange(0, 10, 1, 1, TimeUnit.SECONDS)
//                .take(3, TimeUnit.SECONDS).subscribe(System.out::println);
//        System.out.println("----------------------------");
//        CountDownLatch finish = new CountDownLatch(1);
////        Observable.range(0, 4).skip(2)
////                .subscribe(System.out::println, error -> finish.countDown(), finish::countDown);
//        Observable.interval(1, TimeUnit.SECONDS)
//                .skip(2, TimeUnit.SECONDS)
//                .subscribe(System.out::println, error -> finish.countDown(), finish::countDown);
//
//        Observable.just(1, 2, 3, 4, 5)
//        .elementAt(3).subscribe(System.out::println);
//        Observable.just(1, 2, 3, 4, 5)
//                .elementAt(1, 100).subscribe(System.out::println);
//        Observable.just(1, 2, 3, 4, 5, 6, 7, 9).ignoreElements()
//                .subscribe(() -> System.out.println("completed"));
//        Observable.just(1, 2, 3, 4, 3, 4, 3, 6, 6, 7)
//                .distinct().subscribe(System.out::println);
//        Observable.just(1, 2, 3, 4, 4, 4, 3, 6, 6, 7)
//                .distinctUntilChanged().subscribe(System.out::println);
//        Observable.range(0, 10).filter(x -> x % 2 == 0)
//                .subscribe(System.out::println);
//        System.out.println("----------------------------");
//        Observable.create(emitter -> {
//            if (emitter.isDisposed()) return;
//            try {
//                for (int i = 0; i <= 10; i++) {
//                    emitter.onNext(i);
//                    Thread.sleep(i * 100);
//                }
//                emitter.onComplete();
//            } catch (Exception e) {
//                emitter.onError(e);
//            }
//        }).debounce(100, TimeUnit.MILLISECONDS)
//                .subscribe(System.out::println,
//                        e -> e.printStackTrace(),
//                        () -> System.out.println("completed"));

//        Observable.just(1, 3, 5, 7, 9).all(a -> a < 10).subscribe(System.out::println);
//        Observable.just(1, 2, 3, 4, 5, 6).contains(31).subscribe(System.out::println);
//        Observable.ambArray(
//                Observable.just(1, 2, 3),
//                Observable.just(4, 5, 6),
//                Observable.just(7, 8, 9))
//                .subscribe(System.out::println);
//        Observable.ambArray(
//                Observable.just(1, 2, 3).delay(20, TimeUnit.MILLISECONDS),
//                Observable.just(4, 5, 6))
//                .subscribe(System.out::println);
//        Observable.empty().defaultIfEmpty(100).subscribe(System.out::println);
//        Observable.sequenceEqual(
//                Observable.just(1, 2, 3),
//                Observable.just(1, 2, 3),
//                (a, b) -> a != b).subscribe(System.out::println);
//
//        Observable.interval(1, TimeUnit.SECONDS).
//                skipUntil(Observable.interval(3, TimeUnit.SECONDS))
//                .subscribe(System.out::println);
//        Observable.interval(500, TimeUnit.MILLISECONDS)
//                .skipWhile(a -> a < 3)
//                .subscribe(System.out::println);
//        Thread.sleep(12000);
//        Observable.interval(400, TimeUnit.MILLISECONDS)
//                .takeUntil(a -> a == 5)
//                .subscribe(System.out::println);
//        Observable.interval(400, TimeUnit.MILLISECONDS)
//                .takeWhile(a -> a <= 5)
//                .subscribe(System.out::println);
//        Thread.sleep(12000);
//        Observable observable1 = Observable.just(1, 2, 3, 4, 5);
//        Observable observable2 = Observable.just(4, 5, 6);
//        Observable.merge(observable1, observable2)
//                .subscribe(System.out::println);
//        System.out.println("----------------------------");
//        Observable.mergeDelayError(observable1, observable2)
//                .subscribe(System.out::println);
//        System.out.println("----------------------------");
//        observable1.mergeWith(observable2)
//                .subscribe(System.out::println);
//        Observable observable1 = Observable.just(1, 2, 3, 4, 5);
//        Observable observable2 = Observable.just(4, 5, 6).delay(3, TimeUnit.SECONDS);
//        Observable.zip(
//                observable1,
//                observable2, (a, b) -> a + "ss" + b )
//                .subscribe(System.out::println);
//        Thread.sleep(4000);
//        Observable observable1 = Observable.just(1, 2, 3, 4, 5, 6);
//        Observable observable2 = Observable.just(4, 5, 6).delay(3, TimeUnit.SECONDS);
//        Observable.combineLatest(observable1, observable2, (a, b) -> {
//            System.out.println(a + "+" + b);
//            return a + "+" + b;
//        }).subscribe(System.out::println);
//        Thread.sleep(4000);
//        Observable observable1 = Observable.just(1, 2, 3);
//        Observable observable2 = Observable.just(4, 5, 6, 7);
//        observable1.join(observable2,
//                a -> Observable.just(String.valueOf(a)).delay(200, TimeUnit.MILLISECONDS),
//                b -> Observable.just(String.valueOf(b)).delay(200, TimeUnit.MILLISECONDS),
//                (a, b) -> a + ":" + b).subscribe(System.out::println);
//        Thread.sleep(4000);
//        Observable.just(1, 2, 3).startWith(100).subscribe(System.out::println);
//        Observable.just(1, 2, 3).startWithArray(200, 300).startWith(101).subscribe(System.out::println);
//        Observable.just(1, 2, 3).startWith(Observable.just(99)).subscribe(System.out::println);
        SimpleDateFormat format = new SimpleDateFormat("HH::mm::ss");
        Observable obs = Observable.interval(1, TimeUnit.SECONDS).take(6);
        ConnectableObservable connectableObservable = obs.replay(4);
        connectableObservable.connect();
        connectableObservable
                .subscribe(x -> System.out.println("subscriber1: " + x + " time:" + format.format(new Date())));
        connectableObservable
                .delaySubscription(3, TimeUnit.SECONDS)
                .subscribe(x -> System.out.println("subscriber2: " + x + " time:" + format.format(new Date())));
        Thread.sleep(12000);


    }
}
